RDD - Transformation Filter

The filter() transformation evaluates a user-defined boolean condition on each element of an existing RDD. It returns a new RDD containing only the elements that return True for that condition.

Key Characteristics

• Subset Selection: The output RDD contains a subset of the parent RDD's records. If all elements fail the condition, the child RDD will be completely empty.
• Narrow Dependency: Like map(), filter() processes elements locally within their existing partitions. No global network shuffle is required.
• Preserves Structures: The output elements retain their original data types and forms.

PySpark Code Examples

Setup Spark Session

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("RDD Transformation Filter") \
    .master("local[*]") \
    .getOrCreate()

sc = spark.sparkContext

Example A: Filtering Numeric Values

Let's filter out odd numbers and keep only the even numbers from a list:

# 1. Create a raw RDD
numbers = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])

# 2. Filter for even values (lazy)
evens = numbers.filter(lambda x: x % 2 == 0)

# 3. Trigger action
print("Even Numbers:", evens.collect())
# Output: Even Numbers: [2, 4, 6, 8, 10]

Example B: Extracting Error Lines from Logs

Let's filter raw logging strings to isolate only ERROR state lines:

# 1. Logs RDD
logs = sc.parallelize([
    "INFO: User login successful",
    "WARN: Low disk space detected",
    "ERROR: Connection timed out",
    "INFO: Database index queried",
    "ERROR: Write operation failed"
])

# 2. Filter lines that start with 'ERROR'
errors = logs.filter(lambda log: log.startswith("ERROR"))

# 3. Fetch results
print("Isolated Errors:")
for err in errors.collect():
    print(f"  {err}")

# Expected Output:
# Isolated Errors:
# ERROR: Connection timed out
# ERROR: Write operation failed

Example C: Complex Conditions

Let's filter a collection of users to select only those who are active AND aged 25 or older:

# 1. RDD of user tuples: (Username, Age, IsActive)
users = sc.parallelize([
    ("Alice", 28, True),
    ("Bob", 19, True),
    ("Charlie", 32, False),
    ("David", 24, True),
    ("Eve", 26, True)
])

# 2. Filter with compound lambda condition
active_adults = users.filter(lambda user: user[1] >= 25 and user[2] is True)

print("Active Adults (>=25):", active_adults.collect())
# Output: Active Adults (>=25): [('Alice', 28, True), ('Eve', 26, True)]